#include <iostream>
#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <signal.h>

#include "Sunnet.h"
using namespace std;

Sunnet* Sunnet::instance;

Sunnet::Sunnet() {
    instance = this;

    pthread_rwlock_init(&servicesLock, NULL); // 读写锁初始化
    pthread_spin_init(&globalLock, PTHREAD_PROCESS_PRIVATE);
    pthread_mutex_init(&sleepMtx, NULL);
    pthread_cond_init(&sleepCond, NULL);
    pthread_rwlock_init(&connsLock, NULL);
}

Sunnet::~Sunnet() {
    pthread_rwlock_destroy(&servicesLock);
    pthread_spin_destroy(&globalLock);
    pthread_mutex_destroy(&sleepMtx);
    pthread_cond_destroy(&sleepCond);
    pthread_rwlock_destroy(&connsLock);
}

shared_ptr<BaseMsg> Sunnet::MakeMsg(uint32_t source, char *buff, int len) {
    auto msg = make_shared<ServiceMsg>();
    msg->source = source;
    msg->type = BaseMsg::TYPE::SERVICE;
    msg->buff = shared_ptr<char>(buff);
    msg->size = len;
    return msg;
}

void Sunnet::Start() {
    cout << "Hello Sunnet" << endl;

    // 客户端退出，服务端还向已关闭的套接字发送消息会收到复位信号（RST）
    // 对收到复位信号的套接字调用write，进程会收到SIGPIPE信号，服务端会莫名终止进程
    // 忽略信号
    signal(SIGPIPE, SIG_IGN); 
    

    StartWorker();
    StartSocket();
}

void Sunnet::StartWorker() {
    for (int i = 0; i < WORKER_NUM; ++i) {
        cout << "Start worker thread: " << i << endl; 
        Worker* worker = new Worker();
        worker->id = i;
        worker->eachNum = 1 << i; 
        
        thread *wt = new thread(*worker);
        workers.push_back(worker);
        workerThreads.push_back(wt);
    }
}

void Sunnet::StartSocket() {
    socketWorker = new SocketWorker();
    socketWorker->Init();
    socketThread = new thread(*socketWorker);
}

int Sunnet::Listen(uint32_t port, uint32_t serviceId) {
    int listenfd;
    if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) <= 0) {
        cout << "socket error" << endl;
        return -1;
    }
    fcntl(listenfd, F_SETFL, O_NONBLOCK);
    struct sockaddr_in addr; 
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    if (bind(listenfd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
        cout << "bind error" << endl;
        return -1;
    }
    if (listen(listenfd, 64) < 0) {
        cout << "listen error" << endl;
        return -1;
    }
    AddConn(listenfd, serviceId, Conn::TYPE::LISTEN); // 添加入conn管理， fd->serviceid
    socketWorker->AddEvent(listenfd); // 添加入epoll监听
    return listenfd;
}

void Sunnet::CloseConn(uint32_t fd) {
    bool succ = RemoveConn(fd); // 删除conn对象
    close(fd); // 关闭套接字
    if (succ) {
        socketWorker->RemoveEvent(fd); // 删除epoll对象监听套接字
    }
}

void Sunnet::ModifyEvent(uint32_t fd, bool epollOut) {
    socketWorker->ModifyEvent(fd, epollOut);
    return ;
}

void Sunnet::Wait() {
    if (workerThreads[0]) {
        workerThreads[0]->join();
    }
}

void Sunnet::Send(uint32_t toId, shared_ptr<BaseMsg> msg) {
    shared_ptr<Service> toSrv = GetService(toId);
    if (!toSrv) {
        cout << "Send fail, toSrv not exist toId: " << toId << endl;
        return ;
    }
    toSrv->PushMsg(msg);

    // 检查并放入全局队列
    bool hasPush = false; // 是否执行了插入全局队列操作
    pthread_spin_lock(&toSrv->inGlobalLock);
    {
        /* 
            先给inGlobalLock 加锁，在给全局队列GlobalQueue 加锁
            全局队列更影响性能，临界区必须很小
        */
        if (!toSrv->inGlobal) {
            PushGlobalQueue(toSrv);
            // toSrv->SetInGlobal(true); // 锁住了
            toSrv->inGlobal = true;
            hasPush = true;
        }
    }
    pthread_spin_unlock(&toSrv->inGlobalLock);

    // 唤起进程
    if (hasPush) {
        CheckAndWeakUp();
    }
}

void Sunnet::PushGlobalQueue(shared_ptr<Service> srv) {
    pthread_spin_lock(&globalLock);
    {
        globalQueue.push(srv);
        globalLength ++;
    }
    pthread_spin_unlock(&globalLock);
    return ;
}

shared_ptr<Service> Sunnet::PopGlobalQueue() {
    shared_ptr<Service> srv = NULL; 
    pthread_spin_lock(&globalLock);
    {
        if (!globalQueue.empty()) {
            srv = globalQueue.front();
            globalQueue.pop();
            globalLength --;
        }
    }
    pthread_spin_unlock(&globalLock);
    return srv;
}


uint32_t Sunnet::NewService(shared_ptr<string> type) {
    auto srv = make_shared<Service>(); 
    srv->type = type;
    pthread_rwlock_wrlock(&servicesLock);
    {
        srv->id = maxId;
        maxId ++;
        services.emplace(srv->id, srv);
    }
    pthread_rwlock_unlock(&servicesLock);
    srv->OnInit(); // 服务初始化
    return srv->id;
}

/*
    只能由服务在自己的消息处理函数中调用KillService删除自己
    因为没对srv->OnExit() 和 isExiting 加锁，不具备线程安全性
*/
void Sunnet::KillService(uint32_t id) {
    shared_ptr<Service> srv = GetService(id);
    if (!srv) {
        return ;
    }
    srv->OnExit();
    srv->isExiting = true;
    pthread_rwlock_wrlock(&servicesLock);
    {
        services.erase(id);
    }
    pthread_rwlock_unlock(&servicesLock);
    return ;
}

shared_ptr<Service> Sunnet::GetService(uint32_t id) {
    shared_ptr<Service> srv = NULL; 
    pthread_rwlock_rdlock(&servicesLock);
    {
        unordered_map<uint32_t, shared_ptr<Service>>::iterator it = services.find(id);
        if (it != services.end()) {
            srv = it->second;
        }
    }
    pthread_rwlock_unlock(&servicesLock);
    return srv;
}

void Sunnet::CheckAndWeakUp() {
    // unsafe 
    // 错误唤醒开销小于加锁
    if (sleepCount == 0) {
        return ;
    }
    if (WORKER_NUM - sleepCount <= globalLength) {
        cout << "weakup" << endl; 
        pthread_cond_signal(&sleepCond);
    }
}

void Sunnet::WorkerWait() {
    pthread_mutex_lock(&sleepMtx);
    sleepCount ++;
    pthread_cond_wait(&sleepCond, &sleepMtx);
    sleepCount --; 
    pthread_mutex_unlock(&sleepMtx);
}


int Sunnet::AddConn(int fd, uint32_t id, Conn::TYPE type) {
    auto conn = make_shared<Conn>(); 
    conn->fd = fd; 
    conn->type = type;
    conn->serviceId = id;
    pthread_rwlock_wrlock(&connsLock);
    {
        conns.emplace(fd, conn);
    }
    pthread_rwlock_unlock(&connsLock);
    return fd;
}

shared_ptr<Conn> Sunnet::GetConn(int fd) {
    shared_ptr<Conn> conn = NULL; 
    pthread_rwlock_rdlock(&connsLock);
    {
        unordered_map<uint32_t, shared_ptr<Conn>>::iterator it = conns.find(fd);
        if (it != conns.end()) {
            conn = it->second;
        }
    }
    pthread_rwlock_unlock(&connsLock);
    return conn;
}

bool Sunnet::RemoveConn(int fd) {
    int res;
    pthread_rwlock_wrlock(&connsLock);
    {
        res = conns.erase(fd);
    }
    pthread_rwlock_unlock(&connsLock);
    return res == 1;
}
